package com.hudson.broker.handler.process;

import com.hudson.broker.handler.Process;
import com.hudson.broker.monitor.BizMonitor;
import com.hudson.broker.monitor.ThreadPoolHelper;
import com.hudson.broker.utils.StoreUtil;
import com.hudson.codec.domain.Message;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.Duration;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class RelayProduceProcesser implements Process {

    private static final Logger logger = LoggerFactory.getLogger(RelayProduceProcesser.class);

    @Override
    public RemotingCommand process(ChannelHandlerContext context, RemotingCommand remotingCommand) {
        Message message = Message.deserialize(remotingCommand.getBody());

        Map<String, String> propertiesMap = message.getProperties();
        Duration duration = JSONUtil.fromJson(propertiesMap.get(Duration.class.getSimpleName()), Duration.class);

        Message finalMessage = message;

        ThreadPoolHelper.WHEEL_TIMER.newTimeout(timeout -> {
            // 执行具体的中继任务
            StoreUtil.putMessage(finalMessage);
            logger.debug("延时消息{}延迟{}{}后执行", remotingCommand.getOpaque(), duration.getTime(), duration.getUnit());
        }, duration.getTime(), duration.getUnit());

        BizMonitor.PRODUCE_SPEED.getAndIncrement();


        return RemotingCommand.builder()
                .type(MessageTypeEnum.SEND_MESSAGE_RESP.getType())
                .opaque(remotingCommand.getOpaque())
                .build();
    }
}
